package pulsar.schema.json;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;

/**
 * 
 * @version 1.0
 * @date 2021/3/24 17:27
 */
public class MyConsumer {

    static PulsarClient client;
    static Consumer consumer;

    static {
        try {
            client = PulsarClient.builder()
                        .serviceUrl("pulsar://192.168.75.237:6650")
                        .build();
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
    }

    static void subscribe() throws PulsarClientException {
        consumer = client.newConsumer()
                .topic("my-topic-json")
                .subscriptionName("my-subscription")
                .subscribe();
        while (true) {
            Message msg = consumer.receive();
            try {
                // Do something with the message
                System.out.printf("Message received: %s", new String(msg.getData()));

                // Acknowledge the message so that it can be deleted by the message broker
                consumer.acknowledge(msg);
            } catch (Exception e) {
                e.printStackTrace();
                // Message failed to process, redeliver later
                consumer.negativeAcknowledge(msg);
            }
        }
    }

    public static void main(String[] args) throws PulsarClientException {
        subscribe();

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                consumer.close();
                client.close();
            } catch (PulsarClientException e) {
                e.printStackTrace();
            }
        }));
    }
}
